package com.recsmile.rpc.provider.net.codec;

import com.recsmile.rpc.bo.RpcMessageRequest;
import com.recsmile.rpc.bo.RpcMessageResponse;
import com.recsmile.rpc.provider.MessageCallBack;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by Administrator on 2016/8/3.
 */
public class MessageSendHandler extends ChannelInboundHandlerAdapter {
    private ConcurrentHashMap<String, MessageCallBack> callBackConcurrentHashMap = new ConcurrentHashMap<String, MessageCallBack>();

    private volatile Channel channel;
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //super.channelRead(ctx, msg);
        RpcMessageResponse rpcMessageResponse = (RpcMessageResponse)msg;
        MessageCallBack callBack = callBackConcurrentHashMap.get(rpcMessageResponse.getMessageId());
        if(callBack != null){
            callBackConcurrentHashMap.remove(rpcMessageResponse.getMessageId());
            callBack.over(rpcMessageResponse);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);

        channel = ctx.channel();
    }

    public MessageCallBack sendRequest(RpcMessageRequest rpcMessageRequest){
        MessageCallBack callBack = new MessageCallBack();
        callBackConcurrentHashMap.put(rpcMessageRequest.getMessageId(),callBack);
        channel.writeAndFlush(rpcMessageRequest);
        System.out.println("send;"+ rpcMessageRequest.getMessageId());
        return callBack;
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
}
